package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

//@Configuration
public class HotArticlesStreamHandler2 {


    //@Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder) {
        //1、接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2、聚合流式处理
        stream.map((key,value)->{
            //这里key是null值，value是 UpdateArticleMess 序列化对象
            //反序列化
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            //重置 key是文章id value是行为对应分数
            return new KeyValue<>(mess.getArticleId().toString(),mess.getType() + ":" + mess.getAdd());
        })

                //通过文章id（key）分组
                .groupBy((key,value)-> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
        /**
         * 计算分数
         */
                .aggregate(new Initializer<String>() {
                    //初始化方法 每次时间窗口开始工作都会执行一次，也就执行一次
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     * 真正的聚合方法，具体算分的方法
                     * key value 对应上面重置的key value
                     * aggValue： 无参apply方法的返回值（初始化的值）
                     */
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        //先切割初始化的值 相当于得到一个模板 **,**,**
                        String[] valStr = aggValue.split(",");
                        //初始化值
                        int lik = 0,col = 0,vie = 0,com = 0;
                        //遍历集合
                        for (String val : valStr) {
                            //继续切割 *:*
                            String[] split = val.split(":");
                            //判断
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case LIKES:
                                    lik = Integer.parseInt(split[1]);
                                    break;
                                case COLLECTION:
                                    col = Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        //计算分值 切割value **,**,**
                        String[] valAry = value.split(",");
                        for (String val : valAry) {
                            //切割 *:*
                            String[] split = val.split(":");
                            //判断
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case LIKES:
                                    lik += Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com += Integer.parseInt(split[1]);
                                    break;
                                case COLLECTION:
                                    col += Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    vie += Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        //返回值 格式化结果数据
                        return String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d",col,com,lik,vie);
                    }
                })
                //转成kStream
                .toStream()
                //重新收集 配置
                .map((key,value)->{
                    return new KeyValue<>(key.key().toString(),formatValue(key.key().toString(),value.toString()));
                })
                //发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
                //返回对象
                return stream;
    }

    /**
     * 格式化value对象
     * @param articleId
     * @param value
     * @return
     */
    private String formatValue(String articleId, String value) {
        //创建需要返回的对象
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        //切割value（加工过后的）****，****，****
        String[] valStr = value.split(",");
        //遍历*:*
        for (String val : valStr) {
            //继续切割
            String[] split = val.split(":");
            //判断
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
            }
        }
        //返回对象 序列化
        return JSON.toJSONString(mess);
    }
}
